package com.ydl.learning.flink.demo

import com.ydl.learning.flink.demo.mySink.LogSink
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, OutputTag}
import org.apache.flink.util.Collector

/**
 * 在flink处理数据流时，我们经常会遇到这样的情况：
 * 在处理一个数据源时，往往需要将该源中的不同类型的数据做分割处理，如果使用 filter算子对数据源进行筛选分割的话，势必会造成数据流的多次复制，造成不必要的性能浪费；
 * flink中的旁路输出就是将数据流进行分割，而不对流进行复制的一种分流机制。
 * flink的旁路输出的另一个作用就是对延时迟到的数据进行处理，这样就可以不必丢弃迟到的数据。
 *
 * @author ydl
 * @since 2021/5/7
 */
object SideOutPutDemo extends App with Utils {
  val info = getSocketTextStream(9999)
  val outputTag = OutputTag[String]("side-output")
  val input = info._2
  val mainDataStream = input.process(new ProcessFunction[String, String] {
    override def processElement(value: String, ctx: ProcessFunction[String, String]#Context, out: Collector[String]): Unit = {
      //发送数据到主要的输出
      out.collect(value)
      //发送数据到旁路输出
      if(value.length==1){
        ctx.output(outputTag, "sideout" + value)
      }
    }
  })
  //在 DataStream 运算结果上使用 getSideOutput(OutputTag) 方法获取旁路输出流。这将产生一个与旁路输出流结果类型一致的 DataStream
  val sideOutputStream: DataStream[String] = mainDataStream.getSideOutput(outputTag)

  //  addsink等
  mainDataStream.addSink(new LogSink[String]).name("mainData")
  sideOutputStream.addSink(new LogSink[String]).name("sideOutData")
  info._1.execute("side out put")
}
